#include <unistd.h>
#include "service.h"
#include "z_logger.h"
#include "model.h"
#include "aliot.h"
#include "aiot_subdev_api.h"
#include "subdev_private.h"
#include "cJSON.h"
#include "sqlite.h"
#include "value.h"
#include "z_memory.h"
#include "../hisdata/hisdata.h"

#define TAG_HEAD1 0b1010101 //二进制55
#define TAG_HEAD2 0b10101010 //二进制aa
#define ONLINE_JUDGE "ping -c 3 114.114.114.114"

static block_t block;
static int min_tag = 0, max_tag = 0xffff;
static char log_file[MAX_INPUT] = "/tmp/timestamp.log";
long position = 0;
typedef struct his_value_t {
    uint32_t dev_id;
    uint32_t hour;
    uint32_t tag_id;
} his_value_t;
static his_value_t id = {-1, -1, -1};


void hispub_recv_handler(const aiot_subdev_recv_t *packet) {
    switch (packet->type) {
        case AIOT_SUBDEVRECV_TOPO_ADD_REPLY:
        case AIOT_SUBDEVRECV_TOPO_DELETE_REPLY:
        case AIOT_SUBDEVRECV_TOPO_GET_REPLY:
        case AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY:
        case AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY:
        case AIOT_SUBDEVRECV_SUB_REGISTER_REPLY:
        case AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY:
            log_trace("msg_id=%d, code=%d, product=%s, device=%s, message=%s, data=%s",
                      packet->data.generic_reply.msg_id,
                      packet->data.generic_reply.code,
                      packet->data.generic_reply.product_key,
                      packet->data.generic_reply.device_name,
                      packet->data.generic_reply.message,
                      packet->data.generic_reply.data);
            break;
        case AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY:
            log_trace("msg_id=%d, product=%s, device=%s, params=%s",
                      packet->data.generic_notify.msg_id,
                      packet->data.generic_notify.product_key,
                      packet->data.generic_notify.device_name,
                      packet->data.generic_notify.params);
            break;
        default:
            break;
    }
}

void hispub_event_handler_t() {
}

void *hispub_service_init(aliot_service_args_t *args) {
    void *subdev_handle = aiot_subdev_init();

    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_MQTT_HANDLE, args->mqtt_handle);
    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_RECV_HANDLER, hispub_recv_handler);
    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_EVENT_HANDLER, hispub_event_handler_t);
    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_USERDATA, args->seqiot);

    return subdev_handle;
}

static void hispub_service_fini(void **handle) {
    aiot_subdev_deinit(handle);
}

static int aiot_login_message(subdev_handle_t *subdev_handle, device_t *device) {
    //如果被初始化则直接返回0，防止多次登录报错，否则进行登录操作
    if (device->_stat.initialized) return 0;
    aiot_subdev_dev_t devs;
    devs.device_name = device->device_key;
    devs.device_secret = device->device_secret;
    devs.product_key = device->_product->product_key;
    devs.product_secret = device->_product->product_secret;
    int res = aiot_subdev_send_batch_login(subdev_handle, &devs, 1);
    if (res < 0) {
        return res;
    }
    device->_stat.initialized = true;
    char topic[MAX_INPUT];
    sprintf(topic, "/%s/%s/user/get", device->_product->product_key, device->device_key);
    res = aiot_mqtt_sub(subdev_handle->mqtt_handle, topic, NULL, 1, NULL);
    sleep(5);
    return res;
}

static char *payload_make(const float *value, char *topic, time_t t, thing_model_t *thing_model, device_t *device) {
    char metric[MAX_INPUT];
    char time_char[MAX_INPUT];
    sprintf(metric, "%s", thing_model->metric);
    sprintf(topic, "/sys/%s/%s/thing/event/property/history/post",
            device->_product->product_key, device->device_key);
    cJSON *metric_json = cJSON_CreateObject();
    cJSON *params_json = cJSON_CreateObject();
    cJSON *payload_json = cJSON_CreateObject();
    //测点属性
    cJSON *properties_json = cJSON_CreateObject();
    //身份认证json
    cJSON *identity_json = cJSON_CreateObject();
    cJSON_AddStringToObject(identity_json, "productKey", device->_product->product_key);
    cJSON_AddStringToObject(identity_json, "deviceName", device->device_key);
    float tag_value = *value;
    cJSON_AddNumberToObject(metric_json, "value", tag_value);
    sprintf(time_char, "%ld", t);
    cJSON_AddNumberToObject(metric_json, "time", atof(time_char) * 1000);
    cJSON_AddItemToObject(properties_json, metric, metric_json);
    cJSON_AddItemToObject(params_json, "identity", identity_json);
    cJSON_AddItemToObject(params_json, "properties", properties_json);
    cJSON_AddItemToObject(payload_json, "params", params_json);
    return cJSON_PrintUnformatted(payload_json);
}

static bool
aiot_send_message(subdev_handle_t *subdev_handle, device_t *device, const float *value, thing_model_t *thing_model,
                  time_t t) {
    char topic[MAX_INPUT];
    if (strlen(thing_model->metric) == 0) {
        log_error("product=%s,seg=%s,tag=%s的物模型表中，未配置对应物模型metric数据!", thing_model->product, thing_model->segment,
                  thing_model->super.id);
        goto end;
    }
    //子设备上线
    int res = aiot_login_message(subdev_handle, device);
    if (res < 0) goto end;

    char *payload = payload_make(value, topic, t, thing_model, device);
    res = aiot_mqtt_pub(subdev_handle->mqtt_handle, topic, (uint8_t *) payload, strlen(payload),
                        0);
    if (res == 0) {
        log_trace("his data pub success!");
        return true;
    } else {
        end:
        log_error("his data send failed!");
        return false;
    }
}

//循环获取信息
void cycle_get_info(void *p, hdb_header_t *hdb_header, vector_t *devices, vector_t *products,
                    subdev_handle_t *subdev_handle, time_t time) {
    int send_num = 0;
    int num_tag = hdb_header->n_tags;
    uint32_t *timestamp = &block.head.timestamp;
    for (int i = 0; i < num_tag; i++) {
        tag_header_t *tag_header = (tag_header_t *) p;
        p += sizeof(tag_header_t);
        //读取value
        float *value = p;
        p += sizeof(float);
        if (tag_header->has_timestamp) {
            timestamp = p;
            p += sizeof(uint32_t);
        }
        time_t t = *timestamp;
        //在传入时间time之前的时间不参与补招
        if (t < time) continue;
        struct tm *log_time = localtime(&t);
        if (-1 != id.dev_id && hdb_header->dev_id != id.dev_id) continue;
        if (-1 != id.hour && log_time->tm_hour != id.hour) continue;
        if (tag_header->tag_id < min_tag || tag_header->tag_id > max_tag) continue;
        for (int y = 0; y < vector_size(devices); ++y) {
            device_t *device = vector_get(devices, y);
            if (device->super.id != hdb_header->dev_id) continue;
            for (int z = 0; z < vector_size(products); ++z) {
                product_t *product = vector_get(products, z);
                if (product->super.id != device->product) continue;
                device->_product = product;
                for (int k = 0; k < sizeof(device->_product->thing_model) /
                                    sizeof(device->_product->thing_model[0]); ++k) {
                    vector_t *seg_list = &device->_product->thing_model[k];
                    for (int l = 0; l < vector_size(seg_list); ++l) {
                        thing_model_t *thing_model = vector_get(seg_list, l);
                        int tagId = thing_model_tag_id(thing_model);
                        if (tag_header->tag_id != tagId) continue;
                        //todo 防止阿里云限流429报错
                        bool flag = aiot_send_message(subdev_handle, device, value, thing_model, t);
                        if (flag) {
                            send_num++;
                        }
                        if (send_num % 9 == 0) {
                            sleep(1);
                        }
                    }
                }
            }
        }
    }
}

//分析his文件内容
void
analysis_file(char data_buf[], vector_t *devices, vector_t *products, subdev_handle_t *subdev_handle, time_t time) {
    void *p;
    p = data_buf;
    hdb_header_t *hdb_header = (hdb_header_t *) p;
    p += sizeof(hdb_header_t);
    //循环num_tag个tag_header和具体的tag_value
    cycle_get_info(p, hdb_header, devices, products, subdev_handle, time);
}

void block_read(FILE *file, vector_t *devices, vector_t *products, subdev_handle_t *subdev_handle, time_t time) {
    //读取头部
    while (0 != fread(&block.head, sizeof(block.head), 1, file)) {
        uint16_t data_length = block.head.len;
        char data_buf[data_length];
        bzero(data_buf, sizeof(data_buf));
        //读取数据部分
        fread(data_buf, data_length, 1, file);
        //读取尾部
        fread(&block.tail, sizeof(block.tail), 1, file);
        //判断存储是否有问题
        if (TAG_HEAD1 != block.tail.eof[1] && TAG_HEAD2 != block.tail.eof[0]) {
            log_error("数据块尾部错误！文件存在问题");
            //轮询偏移值，去寻找最近的一个blockhead开头的
            int cycle = 1;
            fseek(file, position + cycle, SEEK_SET);
            bzero(data_buf, sizeof(data_buf));
            while (0 != fread(&block.head, sizeof(block.head), 1, file)) {
                if (TAG_HEAD1 == block.head.bof[0] && TAG_HEAD2 == block.head.bof[1]) {
                    fseek(file, position + cycle, SEEK_SET);
                    position = ftell(file);
                    block_read(file, devices, products, subdev_handle, time);
                }
                fseek(file, position + ++cycle, SEEK_SET);
            }
        } else {
            analysis_file(data_buf, devices, products, subdev_handle, time);
        }
    }
}

void his_pub_func(char *file_path, vector_t *devices, vector_t *products, subdev_handle_t *subdev_handle, time_t time) {
    FILE *file = fopen(file_path, "rb");
    block_read(file, devices, products, subdev_handle, time);
    fclose(file);
}

//根据最后一次上传数据成功后，获取所保存的时间戳
static long time_load(char *log) {
    FILE *file = fopen(log, "r+");
    if (NULL == file) {
        return -1;
    }
    long size = sizeof(time_t);
    fseek(file, -size, SEEK_END);
    long timestamp;
    fread(&timestamp, size, 1, file);
    fclose(file);
    return timestamp;
}

//根据记录时间寻找对应记录文件
void find_his_file(char *filename, time_t log_time) {
    struct tm _tm;
    localtime_r(&log_time, &_tm);
    sprintf(filename, "/usr/data/hdb%04d%02d%02d.db", _tm.tm_year + 1900, _tm.tm_mon + 1, _tm.tm_mday);
    log_info("filename %s's hisdata will be upload", filename);
}

static void *hispub_service_start(void *handle) {
    bool offline = false;
    while (running) {
        //判断离线情况，如果离线则进入下一个循环，否则空转等待
        time_t cur = time(NULL);
        if (0 == cur % 60 && 0 != system(ONLINE_JUDGE)) offline = true;
        while (running && offline) {
            //离线后未上线则继续循环等待，直至上线
            if (0 == cur % 60 && 0 != system(ONLINE_JUDGE)) {
                continue;
            }
            //获取记录时间戳
            time_t log_time = time_load(log_file);
            if (log_time < 0) {
                goto next;
            }
            //判断时间与当前时间是否间隔8h
            time_t now = time(NULL);
            if ((now - log_time) / 3600 > 8)
                goto next;
            log_trace("hisdata start to pub!");
            char filename[NAME_MAX];
            find_his_file(filename, log_time);
            subdev_handle_t *subdev_handle = handle;
            sqlite3 *db = NULL;
            char buf[255];
            if (get_conf_file(buf, sizeof(buf)) >= 0) {
                db = sqlite_open(buf);
            }
            vector_t devices = vector_initializer;
            read_stmt_all_devices(db, &devices);
            vector_t product = vector_initializer;
            read_stmt_products(db, &product);
            for (int i = 0; i < vector_size(&product); ++i) {
                product_t *p = vector_get(&product, i);
                read_stmt_thing_models(db, p);
            }
            //todo 文件位置和判断文件读取方式待定
            his_pub_func(filename, &devices, &product, subdev_handle, log_time);
            usleep(1000 * 1000);

            vector_foreach(&devices, (vector_foreach_callback) device_free);
            vector_clear(&devices);
            vector_foreach(&product, (vector_foreach_callback) product_free);
            vector_clear(&product);

            if (NULL != db) {
                sqlite3_close(db);
            }
            next:
            offline = false;
        }
    }
    return NULL;
}

static aliot_service_t hispub_service = {
        .name = SUBDEV_MODULE_NAME,
        .init = hispub_service_init,
        .fini = hispub_service_fini,
        .start = hispub_service_start,
        .stop = NULL,
};

aliot_service_t *hispub_service_get() {
    return &hispub_service;
}


